package ru.mipt.victator.pagerank;

import java.io.IOException;
import java.util.Properties;

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;

import ru.mipt.victator.pagerank.finalization.Finalization;
import ru.mipt.victator.pagerank.index.IndexJob;
import ru.mipt.victator.pagerank.rank.RankJob;

public class PageRank
{
	private String[] PropertiesList = {
			//Input
			"InputData.PageList.File",			//00
			"InputData.StartPageRank.Float",	//01
			"InputData.Wiki.Dir",				//02
			//Output
			"OutputData.PageRank.Dir",			//03
			"OutputData.Index.Dir",				//04
			//Parameters
			"IndexJob.MapTasks.Int",			//05
			"IndexJob.ReduceTasks.Int",			//06
			"RankJob.MapTasks.Int",				//07
			"RankJob.ReduceTasks.Int",				//08
			"RankJob.Iterations.Int",				//09
			"RankJob.CalculationsParameter.Float"	//10
			};
	private boolean XMLconfig = true;
	private Path ConfigPath = null;
	private Properties Config = null;
	
	private FileSystem FS = null;
	
	public void BlankConfig(Path file) throws IOException
	{
		FSDataOutputStream FSdos = FS.create(file);
		Properties newConfig = new Properties();
		for(int i=0; i<PropertiesList.length; i++)
			newConfig.setProperty(PropertiesList[i],"value");
		if(XMLconfig)
			newConfig.storeToXML(FSdos, "Blank Configuration File for PageRank Calculating Process by Victator", "UTF-8");
		else
			newConfig.store(FSdos, "Blank Configuration File for PageRank Calculating Process by Victator");
		FSdos.close();
	}
	
	public void PrintHelp()
	{
		System.out.println("Usage:");
		System.out.println(	"pagerank -c <config>\n" +
							"      or --config <config>        to start data processing configured by <config>");
		System.out.println(	"pagerank -bc <config>\n" +
						   	"      or --blankconfig <config>   to create blank config file <config>");
		System.out.println(	"pagerank -h,--help                to print this help");
	}
	
	private boolean ParseArgs(String[] args) throws IOException
	{
		for(int i=0; i<args.length; i++)
			try
			{
				if ((args[i].equals("-bc"))||(args[i].equals("--blank_config")))
				{
					BlankConfig(new Path(args[++i]));
					return false;
				}
				if ((args[i].equals("-c"))||(args[i].equals("--config")))
				{
					ConfigPath = new Path(args[++i]);
					if (!FS.exists(ConfigPath))
						System.err.println("ERROR: Reqired ConfigPath (" + ConfigPath.getName() + ") doesn't exist");
					if (!FS.isFile(ConfigPath))
						System.err.println("ERROR: Reqired ConfigPath (" + ConfigPath.getName() + ") isn't file");
				}
				if ((args[i].equals("-h"))||(args[i].equals("--help")))
				{
					PrintHelp();
					return false;
				}
				
			}
			catch (ArrayIndexOutOfBoundsException except)
			{
				System.err.println("ERROR: Required parameter missing from " + args[--i]);
				return false;
			}
		return true;
	}
	
	private boolean TestNotNull(int i, String path)
	{
		if (path == null)
		{
		System.err.println("ERROR: " + PropertiesList[i] + "can't be \"null\"");
		return false;
		}
		else return true;
	}
	
	private boolean TestExists(int i, String path) throws IOException
	{
		if (!FS.exists(new Path(path)))
		{
		System.err.println("ERROR: " + PropertiesList[i] + " doesn't exist");
		return false;
		}
		else return true;
	}
	
	private boolean TestFile(int i, String path) throws IOException
	{
		if (!FS.isFile(new Path(path)))
		{
		System.err.println("ERROR: " + PropertiesList[i] + " must be a file");
		return false;
		}
		else return true;
	}
	
	private boolean TestFloat(int i, String path)
	{
		try
		{
			Float.parseFloat(path);
		}
		catch (NumberFormatException e)
		{
			System.err.println("ERROR: " + PropertiesList[i] + " isn't float");
			return false;
		}
		return true;
	}
	
	private boolean TestInt(int i, String path)
	{
		try
		{
			if (Integer.parseInt(path) <1)
			{
				System.err.println("ERROR: " + PropertiesList[i] + " must be positive");
				return false;
			}
		}
		catch (NumberFormatException e)
		{
			System.err.println("ERROR: " + PropertiesList[i] + " isn't int");
			return false;
		}
		return true;
	}
	
	
	private boolean TestConfig() throws IOException
	{
		String prop;
		//[0]="InputData.PageList.File"
		prop = Config.getProperty(PropertiesList[0]);
		System.out.println(PropertiesList[0] + " = " + prop);
		if (!TestNotNull(0,prop)) return false;
		if (!TestExists(0,prop)) return false;
		if (!TestFile(0,prop)) return false;
		//[1]="InputData.StartPageRank.Float"
		prop = Config.getProperty(PropertiesList[1]);
		System.out.println(PropertiesList[1] + " = " + prop);
		if (!TestNotNull(1,prop)) return false;
		if (!TestFloat(1,prop)) return false;
		//[2]="InputData.Wiki.Dir"
		prop = Config.getProperty(PropertiesList[2]);
		System.out.println(PropertiesList[2] + " = " + prop);
		if (!TestNotNull(2,prop)) return false;
		//[3]="OutputData.PageRank.Dir"
		prop = Config.getProperty(PropertiesList[3]);
		System.out.println(PropertiesList[3] + " = " + prop);
		if (!TestNotNull(3,prop)) return false;
		//[4]="OutputData.Index.Dir"
		prop = Config.getProperty(PropertiesList[4]);
		System.out.println(PropertiesList[4] + " = " + prop);
		if (!TestNotNull(4,prop)) return false;
		//5,6,7,8,9
		for(int i=5; i<10; i++)
		{
			prop = Config.getProperty(PropertiesList[i]);
			System.out.println(PropertiesList[i] + " = " + prop);
			if (!TestNotNull(i,prop)) return false;
			if (!TestInt(i,prop)) return false;
		}
		//[10]="RankJob.CalculationsParameter.Float"
		prop = Config.getProperty(PropertiesList[10]);
		System.out.println(PropertiesList[10] + " = " + prop);
		if (!TestNotNull(10,prop)) return false;
		if (!TestFloat(10,prop)) return false;
		return true;
	}
	
	public boolean Configuration(String[] args) throws IOException
	{
		System.out.println("Configuring data process");
		FS = FileSystem.get(new JobConf());
		if (!ParseArgs(args))
			{
				System.out.println("Configuring failed");
				return false;
			}
		
		Config = new Properties();
		try{
			
		
		if (XMLconfig)
			Config.loadFromXML( FS.open(ConfigPath) );
		else
			Config.load( FS.open(ConfigPath) );
		
		}
		catch(IOException e)
		{
			System.err.println(e.getMessage());
		}
		if (!TestConfig())
		{
			System.out.println("Configuring failed");
			return false;
		}
		
		System.out.println("Configuring completed successfully");
		return true;		
	}
	
	private void runPreProcessing() throws IOException
	{
		PreProcessing preProcessing = new PreProcessing();
		preProcessing.config(FS, Config);
		 System.out.println("Stage 0 - PreProcessing");
		 long time = System.currentTimeMillis();
		int c = preProcessing.process();
		 time = System.currentTimeMillis() - time;
		 System.out.println("Stage 0 (" + c + " records ) - Ok in " + time/60000 + ":" + (time/1000.0)%60 + " [min:sec]");
	}
	
	private void runIndexJob() throws IOException
	{
		 System.out.println("Stage 1 - IndexJob");
		 long time = System.currentTimeMillis();
		JobConf conf = IndexJob.config(Config);
		JobClient.runJob(conf);
		 time = System.currentTimeMillis() - time;
		 System.out.println("Stage 1 - Ok in " + time/60000 + ":" + (time/1000.0)%60 + " [min:sec]");
	}
	
	private void runRankJobs() throws IOException
	{
		 System.out.println("Stage 2 - RankJob");
		 long time = System.currentTimeMillis();
		 
		int iter = Integer.parseInt( Config.getProperty("RankJob.Iterations.Int") );
		for(int i=0; i<iter; i++)
		{
			System.out.println("Stage 2." + i + " - RankJob " + i);
			JobConf conf = RankJob.config(i,Config);
			JobClient.runJob(conf);
		}
		 time = System.currentTimeMillis() - time;
		 System.out.println("Stage 2 - Ok in " + time/60000 + ":" + (time/1000.0)%60 + " [min:sec]");	
	}
	
	private void runFinalization() throws IOException
	{
		Finalization finalization = new Finalization();
		finalization.config(FS, Config);
		 System.out.println("Stage 3 - Finalization");
		 long time = System.currentTimeMillis();
		finalization.process();
		 time = System.currentTimeMillis() - time;
		 System.out.println("Stage 3 - Ok in " + time/60000 + ":" + (time/1000.0)%60 + " [min:sec]");
	}
	
	public void Run() throws IOException
	{
		 System.out.println("PageRank have been started");
		 long time = System.currentTimeMillis();
		runPreProcessing();
		runIndexJob();
		runRankJobs();
		runFinalization();
		 time = System.currentTimeMillis() - time;
		 System.out.println("PageRank have been completed in " + time/60000 + ":" + (time/1000.0)%60 + " [min:sec]");
	}
	
	
	public static void main(String[] args) throws IOException
	{
		PageRank PR = new PageRank();
		if (PR.Configuration(args))
			PR.Run();
	}
}

